package com.study.flink.example.hotitem

import java.util.Properties

import org.apache.commons.lang3.time.FastDateFormat
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
  * 实时热门商品
  * 每5分钟统计最近1小时访问量Top 3 的商品，最大时延5秒
  *
  * @author stephen
  * @date 2019-07-23 14:23
  */
// 原始数据样例类
case class UserBehavior(userId: Long, itemId: Long, categoryId: Long, behavior: String, timestamp: Long)

// 中间聚合结果样例类
case class ItemViewCount(itemId: Long, windowStart: Long, windowEnd: Long, count: Long)

object HotItems {

  def main(args: Array[String]): Unit = {
    // 1 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 按照EventTime进行处理
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 为了打印到控制台的结果不乱序，我们配置全局的并发为1，改变并发对结果正确性没有影响
    env.setParallelism(1)
    env.getConfig.disableSysoutLogging()

    // 2 获取输入数据
    import org.apache.flink.api.scala._
    val properties: Properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")
    //val dataStream = env.addSource(new FlinkKafkaConsumer[String]("behavior", new SimpleStringSchema(), properties))

    val dataStream = env.readTextFile("/Users/stephen/Documents/Code/BigdataWork/bigdata-study/study-flink/src/main/resource/UserBehavior.csv")

    // 3 Transformation
    val resultStream = dataStream
      // 字段转换
      .map(line => {
      val arr = line.split(",")
      UserBehavior(arr(0).trim.toLong, arr(1).trim.toLong, arr(2).trim.toLong, arr(3).trim, arr(4).trim.toLong)
    })
      // 采用周期性生成水印，最大乱序时间为5秒
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[UserBehavior](Time.seconds(5)) {
      override def extractTimestamp(element: UserBehavior): Long = element.timestamp * 1000
    })
      // 过滤出PV
      .filter(_.behavior == "pv")
      // 按照商品ID分组
      .keyBy(_.itemId)
      // 每5分钟处理最近1小时数据
      .timeWindow(Time.hours(1), Time.minutes(5))
      // 根据商品ID聚合点击次数
      .aggregate(new CountAgg(), new WindowResultFunction())
      // 按照窗口关闭时间分组
      .keyBy(_.windowEnd)
      // 根据窗口关闭时间聚合点击次数，并取得点击数前三的商品ID
      .process(new TopNHotItems(3))

    // 4 输出
    resultStream.print()

    // 5 启动任务
    env.execute("Hot Item Demo")
  }
}

/**
  * 预聚合
  *
  * IN：输入数据类型
  * ACC：累加器数据类型
  * OUT：输出数据类型
  */
class CountAgg extends AggregateFunction[UserBehavior, Long, Long] {
  override def createAccumulator(): Long = 0L

  override def add(value: UserBehavior, accumulator: Long): Long = accumulator + 1

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b
}

/**
  * 窗口关闭时的操作
  *
  * IN：预聚合时的数据类型，这里是Long
  * OUT：输出数据类型，这里使用ItemViewCount封装
  * KEY：分组字段的数据类型，分组使用的是itemId，所以是Long
  * W：窗口类型，这里用的是TimeWindow
  */
class WindowResultFunction extends WindowFunction[Long, ItemViewCount, Long, TimeWindow] {

  override def apply(key: Long, window: TimeWindow, input: Iterable[Long], out: Collector[ItemViewCount]): Unit = {
    val itemId = key
    val windowStart = window.getStart
    val windowEnd = window.getEnd
    val count = input.iterator.next()
    out.collect(ItemViewCount(itemId, windowStart, windowEnd, count))
  }
}

/**
  * 获取Top N商品
  * K：分组字段的数据类型，这里分组用的是windowEnd，所以是Long
  * I：输入数据类型，这是ItemViewCount
  * O：输出数据类型，这是是String
  *
  * @param topN
  */
class TopNHotItems(topN: Int) extends KeyedProcessFunction[Long, ItemViewCount, String] {

  // 用于存储商品与点击数的状态，待收齐同一个窗口的数据后，再触发 TopN 计算
  private var itemState: ListState[ItemViewCount] = _
  // 日期输出格式
  private val format: FastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss.SSS")

  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    // 当从故障中恢复，需要重新获取当前环境中的状态
    val itemStateDesc = new ListStateDescriptor[ItemViewCount]("item-state", classOf[ItemViewCount])
    itemState = getRuntimeContext.getListState(itemStateDesc)
  }

  override def processElement(value: ItemViewCount, ctx: KeyedProcessFunction[Long, ItemViewCount, String]#Context, out: Collector[String]): Unit = {
    // 存储每一条数据
    itemState.add(value)
    // 注册一个定时器，延迟触发，Flink会自动忽略同一时间的重复注册
    ctx.timerService().registerEventTimeTimer(value.windowEnd + 1)
  }

  /**
    * 定时器触发时的操作，定时器触发时可以认为数据全部到达
    *
    * @param timestamp
    * @param ctx
    * @param out
    */
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, ItemViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
    // 拿到之前暂存的状态数据
    val allItems: ListBuffer[ItemViewCount] = ListBuffer()
    import scala.collection.JavaConversions._
    for (item <- itemState.get()) {
      allItems.add(item)
    }
    // 尽早清楚状态数据
    itemState.clear()

    // 按照count从大到小排序
    val sortedItems = allItems.sortBy(_.count)(Ordering.Long.reverse).take(topN)

    // 将最终结果转换成String
    val result = new StringBuilder
    result.append("==============================================================================================\n")

    result.append("时间：").append(format.format(timestamp-1)).append("\n")
    for (i <- 0 until sortedItems.size) {
      val item = sortedItems.get(i)
      result.append("NO").append(i+1)
      result.append("  商品ID：").append(item.itemId)
      result.append("  浏览次数：").append(item.count)
      result.append("  窗口时间：(").append(format.format(item.windowStart)).append(") <=> (").append(format.format(item.windowEnd)).append(")\n")
    }

    result.append("==============================================================================================\n")

    Thread.sleep(1000)

    out.collect(result.toString())
  }

}